//import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.*;


public class GroupCombineonaGroupedDataSet
{

    public static void main(String[] args) throws Exception
    {



        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String rootPath = new StringBuilder(System.getProperty("user.dir")).toString();
//        DataSet<Tuple1<String>> input = env.readCsvFile("file:///"+rootPath+"/"+"mapper.csv").types(String.class); // The words received as input
        DataSet<String> input = env.readTextFile("file:///"+rootPath+"/"+"mapper.csv");

        DataSet<Tuple2<String, Integer>> combinedWords = input
//                .groupBy(0) // group identical words
                .combineGroup(new GroupCombineFunction<String, Tuple2<String, Integer>>()
                {
                    @Override
                    public void combine(Iterable<String> words, Collector<Tuple2<String, Integer>> out) throws Exception {
                        int count = 0;
                        HashMap strihMap = new HashMap();
                        /*List<HashMap> objects = new ArrayList<>();*/

                        for (String word : words)
                        {
                            if( strihMap.keySet().contains(word)){
                                count =  Integer.valueOf(strihMap.get(word).toString());
                                strihMap.put(word,count+1);
                            }else{
                                strihMap.put(word,1);
                            }
                         /*   objects.add(strihMap);*/
                        }
                        /*for (int i = objects.size()-1; i < 0; i--) {
                            HashMap hashMap = objects.get(i);
                            hashMap.

                        }*/
                        /*Iterator iterator = strihMap.keySet().iterator();
                        while (iterator.hasNext()){
                            Object next = iterator.next();
                            System.out.println("----：" + next);
                            Object o = strihMap.get(strihMap);
                        }*/
                        out.collect(new Tuple2<>(strihMap.toString(), count));

                        // emit tuple with word and count
                    }
                });

        System.out.println("-----------------------输出combinedWords-------------------------------");
        combinedWords.print();





        DataSet<Tuple2<String, Integer>> output = combinedWords.groupBy(0).reduceGroup(new GroupReduceFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>()
        {
            @Override
            public void reduce(Iterable<Tuple2<String, Integer>> words, Collector<Tuple2<String, Integer>> out ) throws Exception
            {
                String key = "";
                int count = 0;
                for (Tuple2<String, Integer> word : words)
                {
                    key=word.f0;
                    System.out.println("----：" + key);
                    count++;
                }

                out.collect(new Tuple2<>(key, count));

            }
        });


        System.out.println("-----------------------输出output------------------------------");
        output.print();
    }
}









